/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.activity.impl;

import org.sabayframework.activity.Activity;
import org.sabayframework.activity.ActivityContext;
import org.sabayframework.channel.Channel;
import org.sabayframework.channel.ram.ConcurrentArrayChannel;
import org.sabayframework.queue.ExecutionQueue;



/**
 * Pipeline is composition of activities that can run in parallel and
 * the result of each activity is passed as the argument of the next, 
 * and the result of the last one is the result of the whole. 
 * 
 * @author Alex Antonau
 * */
public final class Pipeline<I,O> implements Activity<I, O> {
	
	private int channelSize;
	@SuppressWarnings("rawtypes")
	private Activity[] activities;
	private boolean done;

	public Pipeline() {
		super();
	}
	public Pipeline(Activity<?, ?>... activities) {
		this(100, activities);
	}
	public Pipeline(int channelSize, Activity<?, ?>... activities) {
		if(activities == null || activities.length == 0) throw new IllegalArgumentException();
		setActivities(activities);
		setChannelSize(channelSize);
	}
	
	public int getChannelSize() {
		return channelSize;
	}
	public void setChannelSize(int channelSize) {
		this.channelSize = channelSize;
	}
	public Activity<?, ?>[] getActivities() {
		return activities;
	}
	public void setActivities(Activity<?, ?>... activities) {
		this.activities = activities;
	}
	public boolean isDone(ActivityContext<I, O> ctx) {
		return done;
	}

	public boolean isReady(ActivityContext<I, O> ctx) {
		return !done;
	}
	@SuppressWarnings({ "rawtypes", "unchecked" })
	public void execute(ActivityContext<I, O> ctx) throws Exception {
		done= true;
		ExecutionQueue fs = ctx.getExecutionQueue();
		Channel input = ctx.getInput();
		for(int i=0;i<activities.length;++i){
			if(i==activities.length-1){
				fs.submit(activities[i], input, ctx.getOutput());
			}
			else {
				Channel<Object> output = new ConcurrentArrayChannel<Object>(channelSize);
				fs.submit(activities[i], input, output);
				input = output;
			}
		}
	}
	public void beforeExecute(ActivityContext<I, O> ctx) throws Exception {
	}
	public void afterExecute(ActivityContext<I, O> ctx) throws Exception {
	}
	
}
